//package com.light.dp.flink.config;
//
//import com.light.dp.flink.config.dto.*;
//import com.light.dp.flink.dto.DataChangeInfo;
//import com.light.dp.flink.dto.User;
//import com.light.dp.flink.enums.JdbcDriverEnum;
//import com.light.dp.flink.serialization.MysqlDeserialization;
//import com.light.dp.flink.sink.*;
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.configuration.RestOptions;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.Resource;
//import java.util.Arrays;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//import java.util.stream.Collectors;
//
//@Component
//public class ClickHouseFlinkCDCConfig {
//    @Resource
//    private User user;
//
//    @PostConstruct
//    public void init() throws Exception {
//
//        new Thread(() -> {
//            System.out.println(user.getName());
//
//            // 配置 Flink WebUI
//            Configuration configuration = new Configuration();
//            configuration.setInteger(RestOptions.PORT, 8078);
//            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
//            env.setParallelism(2); // 设置全局并行度为 4
//            // 开启检查点机制
//            env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
//
//            // 设置固定延迟重启策略，重试 5 次，每次重试间隔 5 秒
//            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
//                    Integer.MAX_VALUE, // 重试次数
//                    1000 // 重试间隔，单位为毫秒
//            ));
//
//            env.setStateBackend(new FsStateBackend("file:///D:/git/light-data-processing/checkpoints"));
//
//            MysqlMapping collectMapping = getOneToManyMapping();
//
//            // 为每个源表和其对应的目标表创建数据流和 Sink
//            for (SourceJDBCDatabaseConfig sourceConfig : collectMapping.getSourceConfigs()) {
//                // 获取对应的源表集合
//                List<String> sourceTables = sourceConfig.getTableMappings().stream().map(t -> sourceConfig.getDatabase() + "." + t.getSourceTable()).collect(Collectors.toList());
//
//                MySqlSource<DataChangeInfo> source = createMySqlSource(sourceConfig, sourceTables);
//                for (TargetJDBCDatabaseConfig targetConfig : sourceConfig.getTargetConfigs()) {
//                    DataStream<DataChangeInfo> dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
//                            "Source-" + sourceConfig.getHostname() + ":" + sourceConfig.getPort() + "/" + sourceConfig.getDatabase() +
//                                    targetConfig.getJobName());
//
//
//                    dataStream.addSink(new ClickHouseIncrementalSink(targetConfig, sourceConfig.getTableMappings())).name("Mysql To ClickHouse");
//                        dataStream.addSink(new ClickHouseStockSink1(targetConfig, sourceConfig.getTableMappings())).name("Mysql To ClickHouse1");
//                }
//            }
//
//            // 执行作业
//            try {
//
//                env.execute("Multi Source Multi Table to Multi Target Multi Table");
//            } catch (Exception e) {
//                throw new RuntimeException(e);
//            }
//
//        }).start();
//    }
//
//    private static MySqlSource<DataChangeInfo> createMySqlSource(JDBCDatabaseConfig config, List<String> sourceTables) {
//
//        return MySqlSource.<DataChangeInfo>builder()
//                .hostname(config.getHostname())
//                .port(config.getPort())
//                .databaseList(config.getDatabase())
//                .tableList(StringUtils.join(sourceTables, ","))
//                .username(config.getUsername())
//                .password(config.getPassword())
//                // 添加时区配置
////                .serverTimeZone("Asia/Shanghai")
//                .serverTimeZone("UTC")
//                .deserializer(new MysqlDeserialization())
//                .includeSchemaChanges(true)
////                .startupOptions(StartupOptions.initial())
////                .startupOptions(StartupOptions.latest())
//                .startupOptions(StartupOptions.specificOffset("binlog.000040", 507778160))
////                .startupOptions(StartupOptions.timestamp(Timestamp.valueOf("2025-02-21 00:00:00").getTime()))
//                .build();
//    }
//
//
//    /**
//     * 测试一源对多库2025-02-24 17:47:25.020
//     * 结果：多库其中一台宕机后恢复，缺少的数据会重新写入
//     * @return
//     */
//    private MysqlMapping getOneToManyMapping() {
//        // 列映射信息
//        List<ColumnMapping> columnMappings = Arrays.asList(new ColumnMapping("id", "id", true),
//                new ColumnMapping("deleted","deleted"));
//
//        // 列映射信息
////        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("quick_chat_msg", "buffer_quick_chat_msg", null));
//        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("quick_chat_msg", "quick_chat_msg", columnMappings));
//
//
//
//        // 目标库配置
//        List<TargetJDBCDatabaseConfig> targetConfigs = Arrays.asList(
//                new TargetJDBCDatabaseConfig(JdbcDriverEnum.CLICKHOUSE,"192.168.1.72", 8123, "target_db3", "default", "default")
////                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL, "localhost", 3306, "target_db2", "root", "root")
//        );
//
//        // 源库配置
//        List<SourceJDBCDatabaseConfig> sourceConfigs = Arrays.asList(
//                new SourceJDBCDatabaseConfig("192.168.1.72", 3307, "city-supervise-service", "canal", "canal", targetConfigs, tableMappings)
//        );
//
//        return new MysqlMapping(sourceConfigs);
//    }
//
//}
//
